package com.rightsidetech.http.mq;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

@Component
@Slf4j
public class ProducerMsg {
    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup; //生产者

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Value("${apache.rocketmq.tags}")
    private String tags;

    private DefaultMQProducer producer;//默认生产者
    //提供get方法，让外部调用
    public DefaultMQProducer getProducer() {
        return producer;
    }

    /**
     * @PostConstruct在servlet初始化时执行，只执行一次 作用：让当前方法在构造函数之上，在init方法执行之前
     */
    @PostConstruct
    public void defaultMQProducer() {
        //1：创建一个默认的生产者对象--作用用于生成消息
        producer = new DefaultMQProducer(producerGroup);

        //2:绑定生产者和nameserver，就是建立和broker程序的关系
        producer.setNamesrvAddr(namesrvAddr);
        //3:发送消息
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    @Async
    public void sendMq(String command,String terminalNo,String road,String content,int timeLevel){
        Map map = new HashMap();
        map.put("command",command);
        map.put("terminalNo", terminalNo);
        map.put("channelNo", road);
        map.put("msg", content);
        map.put("requestId", UUID.randomUUID().toString().replace("-",""));
        String msg = JSON.toJSONString(map);
        log.info("定时任务, 添加消息队列,command={}, imeiNo={},channelNo={}, msg={},timeLevel={}", command,terminalNo,road, msg, timeLevel);
        Message message = new Message("cworderTopic", tags, msg.getBytes(StandardCharsets.UTF_8));
        //级别 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        message.setDelayTimeLevel(timeLevel);
        try {
            SendResult send = getProducer().send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    @Async //异步调用,下单
    public void sendPalceOrderMq(String orderNo,Integer delayTimeLevel){
        try {
            Message msg = new Message("cworderTopic",tags,orderNo.getBytes("UTF-8"));
            msg.setDelayTimeLevel(delayTimeLevel);//30分钟延时消息
            SendResult sendResult = getProducer().send(msg);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("发送mq消息错误","订单号:" + orderNo);
        }

    }

    public void sendMsgOrderMq(Map<String, String> map, Integer delayTimeLevel) {
        try {
            Message msg = new Message("orderTopic",tags,JSON.toJSONString(map).getBytes("UTF-8"));
            msg.setDelayTimeLevel(delayTimeLevel);//20分钟延时消息
            getProducer().send(msg);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("发送mq消息错误","消息内容:" + JSON.toJSONString(map));
        }


    }
}
